package com.ydlclass.codec;

import com.ydlclass.compress.Compress;
import com.ydlclass.configuration.YrpcConfiguration;
import com.ydlclass.configuration.YrpcConfigurationHolder;
import com.ydlclass.message.MessageFormatConst;
import com.ydlclass.message.MessageType;
import com.ydlclass.message.YrpcMessage;
import com.ydlclass.serialization.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;

/**
 * <p>
 * 自定义协议编码器
 * <p>
 * <pre>
 *   0    1    2    3    4    5    6    7    8    9    10   11   12   13   14   15   16   17   18   19   20   21
 *   +----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+----+
 *   |    magic     |ver |head  len|    full length    | mt | ser|comp|              RequestId                |
 *   +-----+-----+-------+----+----+----+----+-----------+----- ---+--------+----+----+----+----+----+----+---+
 *   |                                                                                                        |
 *   |                                         body                                                           |
 *   |                                                                                                        |
 *   +--------------------------------------------------------------------------------------------------------+
 *   3B magic（魔数）   --> ydl
 *   1B version（版本） --> 01
 *   2B header length  （head的长度，用于扩展）
 *   4B full length（消息长度）
 *   1B messageType（消息类型）
 *   1B serialize（序列化类型）
 *   1B compress（压缩类型）
 *   8B requestId（请求的Id）
 *   body（object类型数据）
 * </pre>
 *
 */
@Slf4j
public class YrpcMessageEncoder extends MessageToByteEncoder<YrpcMessage> {

    @Override
    protected void encode(ChannelHandlerContext ctx, YrpcMessage yrpcMessage, ByteBuf out) {
        // 3B ydl magic code（魔数）
        out.writeBytes(MessageFormatConst.MAGIC);
        // 1B version（版本）
        out.writeByte(MessageFormatConst.VERSION);
        // 2B header长度
        out.writeShort(MessageFormatConst.HEADER_LENGTH);
        // 4B full length（消息长度）. 总长度先空着，后面填。
        out.writerIndex(out.writerIndex() + MessageFormatConst.FULL_LENGTH_LENGTH);
        // 1B messageType（消息类型）
        out.writeByte(yrpcMessage.getMessageType());
        // 1B codec（序列化类型）
        out.writeByte(yrpcMessage.getSerializeType());
        // 1B compress（压缩类型）
        out.writeByte(yrpcMessage.getCompressTye());
        // 8B requestId（请求的Id）
        out.writeLong(yrpcMessage.getRequestId());
        // 写 body，返回 body 长度
        int bodyLength = writeBody(yrpcMessage, out);

        // 记录一下当前的指针
        int writerIndex = out.writerIndex();
        out.writerIndex(MessageFormatConst.MAGIC_LENGTH + MessageFormatConst.VERSION_LENGTH +MessageFormatConst.HEAD_LENGTH);
        // 4B full length（消息长度）
        out.writeInt(MessageFormatConst.HEADER_LENGTH + bodyLength);
        // 写指针复原
        out.writerIndex(writerIndex);
    }

    /**
     * 写 body
     *
     * @return body 长度
     */
    private int writeBody(YrpcMessage yrpcMessage, ByteBuf out) {
        byte messageType = yrpcMessage.getMessageType();
        // 如果是 ping、pong 心跳类型的，没有 body，直接返回头部长度
        if (messageType == MessageType.HEARTBEAT.getValue()) {
            return 0;
        }

        // 序列化
        YrpcConfiguration yrpcConfiguration = YrpcConfigurationHolder.get();
        Serializer serializer = yrpcConfiguration.getSerializer();
        if (serializer == null) {
            throw new IllegalArgumentException("A suitable serializer was not found.");
        }
        byte[] notCompressBytes = serializer.serialize(yrpcMessage.getPayload());
        // 压缩
        Compress compressor = yrpcConfiguration.getCompress();
        if (compressor == null) {
            throw new IllegalArgumentException("A suitable compressor was not found.");
        }
        if (log.isDebugEnabled()){
            log.debug("--> The message[{}] has been serialized,size ---> {}",
                    yrpcMessage.getRequestId(),notCompressBytes.length);
        }
        byte[] compressedBytes = compressor.compress(notCompressBytes);
        if (log.isDebugEnabled()){
            log.debug("--> The message[{}] has been compressed,size ---> {}",
                    yrpcMessage.getRequestId(),compressedBytes.length);
        }
        // 写 body
        out.writeBytes(compressedBytes);
        return compressedBytes.length;
    }

}
